/*
 * Unidata Platform Community Edition
 * Copyright (c) 2013-2020, UNIDATA LLC, All rights reserved.
 * This file is part of the Unidata Platform Community Edition software.
 *
 * Unidata Platform Community Edition is free software: you can redistribute it and/or modify
 * it under the terms of the GNU General Public License as published by
 * the Free Software Foundation, either version 3 of the License, or
 * (at your option) any later version.
 *
 * Unidata Platform Community Edition is distributed in the hope that it will be useful,
 * but WITHOUT ANY WARRANTY; without even the implied warranty of
 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 * GNU General Public License for more details.
 *
 * You should have received a copy of the GNU General Public License
 * along with this program. If not, see <https://www.gnu.org/licenses/>.
 */

package org.unidata.mdm.system.configuration;

import java.nio.charset.StandardCharsets;
import java.util.Arrays;
import java.util.Collections;
import java.util.Map;
import java.util.Objects;
import java.util.Properties;
import java.util.Set;

import javax.sql.DataSource;

import org.apache.camel.management.JmxManagementStrategyFactory;
import org.apache.camel.processor.aggregate.GroupedBodyAggregationStrategy;
import org.apache.commons.beanutils.ConvertUtils;
import org.apache.commons.beanutils.PropertyUtils;
import org.apache.commons.lang3.ArrayUtils;
import org.apache.commons.lang3.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.config.PropertiesFactoryBean;
import org.springframework.context.ApplicationContext;
import org.springframework.context.MessageSource;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.context.support.PropertySourcesPlaceholderConfigurer;
import org.springframework.context.support.ResourceBundleMessageSource;
import org.springframework.core.env.Environment;
import org.springframework.core.env.StandardEnvironment;
import org.springframework.core.io.ClassPathResource;
import org.springframework.jdbc.datasource.SingleConnectionDataSource;
import org.springframework.transaction.PlatformTransactionManager;
import org.springframework.transaction.jta.JtaTransactionManager;
import org.unidata.mdm.system.service.PlatformConfiguration;

import com.hazelcast.config.Config;
import com.hazelcast.config.JoinConfig;
import com.hazelcast.config.MulticastConfig;
import com.hazelcast.config.NetworkConfig;
import com.hazelcast.config.TcpIpConfig;
import com.hazelcast.core.Hazelcast;
import com.hazelcast.core.HazelcastInstance;

import bitronix.tm.TransactionManagerServices;

/**
 * @author Mikhail Mikhailov
 * Root spring context link.
 */
@Configuration
public class SystemConfiguration extends AbstractConfiguration {

    private static final Logger LOGGER = LoggerFactory.getLogger(SystemConfiguration.class);

    private static final ConfigurationId ID = () -> "SYSTEM_CONFIGURATION";

    private ResourceBundleMessageSource systemMessageSource;

    /**
     * Constructor.
     */
    public SystemConfiguration() {
        super();

        // Do it here, because otherwise, we can not control bundles joining
        systemMessageSource = new ResourceBundleMessageSource();
        systemMessageSource.setDefaultEncoding(StandardCharsets.UTF_8.name());
    }

    /**
     * Link to system bundles.
     */
    public ResourceBundleMessageSource getSystemMessageSource() {
        return systemMessageSource;
    }

    /**
     * {@inheritDoc}
     */
    @Override
    protected ConfigurationId getId() {
        return ID;
    }

    public static ApplicationContext getApplicationContext() {
        return CONFIGURED_CONTEXT_MAP.get(ID);
    }
    /**
     * Gets a bean.
     *
     * @param <T>
     * @param beanClass the bean class
     * @return bean
     */
    public static <T> T getBean(Class<T> beanClass) {
        if (CONFIGURED_CONTEXT_MAP.containsKey(ID)) {
            return CONFIGURED_CONTEXT_MAP.get(ID).getBean(beanClass);
        }

        return null;
    }

    /**
     * Gets beans of type.
     *
     * @param <T>
     * @param beanClass the bean class
     * @return bean
     */
    public static <T> Map<String, T> getBeans(Class<T> beanClass) {
        if (CONFIGURED_CONTEXT_MAP.containsKey(ID)) {
            return CONFIGURED_CONTEXT_MAP.get(ID).getBeansOfType(beanClass);
        }

        return Collections.emptyMap();
    }

    @Bean
    public HazelcastInstance hazelcastInstance(@Autowired Environment env) {

        // 0. Top container
        final Config config = new Config()
                .setInstanceName(SystemConfigurationConstants.SYSTEM_APLLICATION_NAME);

        // 1. General network settings
        final NetworkConfig network = config.getNetworkConfig();

        int port  = env.getProperty(SystemConfigurationConstants.PROPERTY_SYSTEM_CACHE_PORT, Integer.class, 5701);
        boolean increment = env.getProperty(SystemConfigurationConstants.PROPERTY_SYSTEM_CACHE_PORT_AUTOINCREMENT, Boolean.class, false);
        boolean reuse = env.getProperty(SystemConfigurationConstants.PROPERTY_SYSTEM_CACHE_REUSE_ADDERSS, Boolean.class, false);

        network.setPort(port);
        network.setPortAutoIncrement(increment);
        network.setReuseAddress(reuse);

        int count = env.getProperty(SystemConfigurationConstants.PROPERTY_SYSTEM_CACHE_PORT_COUNT, Integer.class, -1);
        if (count > 0) {
            network.setPortCount(count);
        }

        String publicAddress = env.getProperty(SystemConfigurationConstants.PROPERTY_SYSTEM_CACHE_PUBLIC_ADDRESS, String.class, StringUtils.EMPTY);
        if (StringUtils.isNotBlank(publicAddress)) {
            network.setPublicAddress(StringUtils.trim(publicAddress));
        }

        String interfaces = env.getProperty(SystemConfigurationConstants.PROPERTY_SYSTEM_CACHE_INTERFACES, String.class, StringUtils.EMPTY);
        if (StringUtils.isNotBlank(interfaces)) {

            String[] split = StringUtils.split(StringUtils.trim(interfaces), ',');
            if (ArrayUtils.isNotEmpty(split)) {

                network.getInterfaces()
                    .setEnabled(true)
                    .setInterfaces(Arrays.asList(split));
            }
        }

        String outboundPorts = env.getProperty(SystemConfigurationConstants.PROPERTY_SYSTEM_CACHE_OUTBOUND_PORTS, String.class, StringUtils.EMPTY);
        if (StringUtils.isNotBlank(outboundPorts)) {
            network.addOutboundPortDefinition(StringUtils.trim(outboundPorts));
        }

        // 2. Join methods
        final JoinConfig join = network.getJoin();

        // 2.1. TCP/IP
        TcpIpConfig tcp = join.getTcpIpConfig();

        boolean tcpEnabled = env.getProperty(SystemConfigurationConstants.PROPERTY_SYSTEM_CACHE_TCP_IP_ENABLED, Boolean.class, false);

        tcp.setEnabled(tcpEnabled);

        String tcpMembers = env.getProperty(SystemConfigurationConstants.PROPERTY_SYSTEM_CACHE_TCP_IP_MEMBERS, String.class, StringUtils.EMPTY);
        if (StringUtils.isNotBlank(tcpMembers)) {
            tcp.addMember(StringUtils.trim(tcpMembers));
        }

        int tcpTimeout = env.getProperty(SystemConfigurationConstants.PROPERTY_SYSTEM_CACHE_TCP_IP_TIMEOUT, Integer.class, -1);
        if (tcpTimeout > 0) {
            tcp.setConnectionTimeoutSeconds(tcpTimeout);
        }

        // 2.2. Multicast
        MulticastConfig multicast = join.getMulticastConfig();

        boolean multicastEnabled = env.getProperty(SystemConfigurationConstants.PROPERTY_SYSTEM_CACHE_MULTICAST_ENABLED, Boolean.class, false);

        multicast.setEnabled(multicastEnabled);

        String multicastGroup = env.getProperty(SystemConfigurationConstants.PROPERTY_SYSTEM_CACHE_MULTICAST_GROUP, String.class, StringUtils.EMPTY);
        if (StringUtils.isNotBlank(multicastGroup)) {
            multicast.setMulticastGroup(StringUtils.trim(multicastGroup));
        }

        int multicastPort  = env.getProperty(SystemConfigurationConstants.PROPERTY_SYSTEM_CACHE_MULTICAST_PORT, Integer.class, -1);
        if (multicastPort > 0) {
            multicast.setMulticastPort(multicastPort);
        }

        int multicastTTL  = env.getProperty(SystemConfigurationConstants.PROPERTY_SYSTEM_CACHE_MULTICAST_TTL, Integer.class, -1);
        if (multicastTTL > 0) {
            multicast.setMulticastTimeToLive(multicastTTL);
        }

        int multicastTimeout  = env.getProperty(SystemConfigurationConstants.PROPERTY_SYSTEM_CACHE_MULTICAST_TIMEOUT, Integer.class, -1);
        if (multicastTimeout > 0) {
            multicast.setMulticastTimeoutSeconds(multicastTimeout);
        }

        String multicastTrusted = env.getProperty(SystemConfigurationConstants.PROPERTY_SYSTEM_CACHE_MULTICAST_TRUSTED, String.class, StringUtils.EMPTY);
        if (StringUtils.isNotBlank(multicastTrusted)) {

            String[] split = StringUtils.split(StringUtils.trim(multicastTrusted), ',');
            if (ArrayUtils.isNotEmpty(split)) {
                multicast.setTrustedInterfaces(Set.of(split));
            }
        }

        return Hazelcast.getOrCreateHazelcastInstance(config);
    }

    @Bean
    public PropertySourcesPlaceholderConfigurer propertySourcesPlaceholderConfigurer() {
        final PropertySourcesPlaceholderConfigurer propertySourcesPlaceholderConfigurer =
                new PropertySourcesPlaceholderConfigurer();
        propertySourcesPlaceholderConfigurer.setEnvironment(new StandardEnvironment());
        return propertySourcesPlaceholderConfigurer;
    }

    @Bean("systemDataSource")
    public DataSource systemDataSource(@Autowired Environment env) {

        // Single connection data source
        String url = env.getProperty(SystemConfigurationConstants.PROPERTY_SYSTEM_DATASOURCE_URL);
        String username = env.getProperty(SystemConfigurationConstants.PROPERTY_SYSTEM_DATASOURCE_USER);
        String password = env.getProperty(SystemConfigurationConstants.PROPERTY_SYSTEM_DATASOURCE_PASSWORD);

        Objects.requireNonNull(url, "System datasource URL cannot be null");

        SingleConnectionDataSource scds = new SingleConnectionDataSource(url, username, password, true);
        scds.setDriverClassName("org.postgresql.Driver");

        return scds;
    }

    @Bean
    public PlatformTransactionManager transactionManager(PlatformConfiguration pc) {

        bitronix.tm.Configuration btmc = TransactionManagerServices.getConfiguration();
        Properties properties = getAllPropertiesWithPrefix(SystemConfigurationConstants.PROPERTY_BITRONIX_PREFIX, true);
        if (Objects.nonNull(properties)) {

            properties.forEach((k, v) -> {

                // Property names differ
                String propName = StringUtils.substringAfterLast(k.toString(), ".");
                if ("maxLogSize".equals(propName)) {
                    propName = "maxLogSizeInMb";
                } else if ("async".equals(propName)) {
                    propName = "asynchronous2Pc";
                } else if ("sync".equals(propName)) {
                    propName = "synchronousJmxRegistration";
                } else if ("userTransactionName".equals(propName)) {
                    propName = "jndiUserTransactionName";
                } else if ("transactionSynchronizationRegistryName".equals(propName)) {
                    propName = "jndiTransactionSynchronizationRegistryName";
                } else if ("configuration".equals(propName)) {
                    propName = "resourceConfigurationFilename";
                }

                try {
                    // Overwrite non-null values only
                    Class<?> targetClazz = PropertyUtils.getPropertyType(btmc, propName);
                    if (targetClazz != null) {

                        String stringVal = Objects.nonNull(v) ? v.toString() : null;
                        if (StringUtils.isBlank(stringVal)) {
                            return;
                        }

                        if (targetClazz == String.class) {
                            PropertyUtils.setProperty(btmc, propName, stringVal);
                        } else {

                            Object targetValue = ConvertUtils.convert(stringVal, targetClazz);
                            if (targetValue != stringVal) {
                                PropertyUtils.setProperty(btmc, propName, targetValue);
                            }
                        }
                    }
                } catch (Exception e) {
                    LOGGER.warn("Cannot wire property value to BITRONIX configuration.", e);
                }
            });
        }

        btmc.setServerId("BTM-" + pc.getNodeId());
        if ("auto".equals(btmc.getJdbcProxyFactoryClass())) {
            btmc.setJdbcProxyFactoryClass("bitronix.tm.resource.jdbc.proxy.JdbcJavaProxyFactory");
        }

        JtaTransactionManager jtm = new bitronix.tm.integration.spring.PlatformTransactionManager();
        jtm.setAllowCustomIsolationLevels(true);

        return jtm;
    }

    @Bean("configuration-sql")
    public PropertiesFactoryBean configurationSql() {
        final PropertiesFactoryBean propertiesFactoryBean = new PropertiesFactoryBean();
        propertiesFactoryBean.setLocation(new ClassPathResource("/db/configuration-sql.xml"));
        return propertiesFactoryBean;
    }

    @Bean("pipelines-sql")
    public PropertiesFactoryBean pipelinesSql() {
        final PropertiesFactoryBean propertiesFactoryBean = new PropertiesFactoryBean();
        propertiesFactoryBean.setLocation(new ClassPathResource("/db/pipelines-sql.xml"));
        return propertiesFactoryBean;
    }

    @Bean
    public MessageSource messageSource() {
        return systemMessageSource;
    }

    @Bean
    public JmxManagementStrategyFactory jmxManagementStrategyFactory() {
        return new JmxManagementStrategyFactory();
    }

    @Bean
    public GroupedBodyAggregationStrategy groupedBodyAggregationStrategy() {
        return new GroupedBodyAggregationStrategy();
    }
}
